#include <map>
#include <cppformat/format.h>

#include "command.hpp"
#include "server.hpp"
#include "client.hpp"
#include "proxy.hpp"
#include "response.hpp"
#include "except/exceptions.hpp"
#include "utils/alg.hpp"
#include "utils/logging.hpp"
#include "syscalls/poll.h"
#include "syscalls/fctl.h"

using namespace cerb;

void Server::on_events(int events)
{
    if (closed()) {
        return;
    }
    if (poll::event_is_hup(events)) {
		//if node suddenly crashed，it won't pass here without RDHUP and it will 
		//throw at _recv_from beginning
		//But when node initially created, it support RDHUP, so it will go here
        return close_conn();
    }
    if (poll::event_is_read(events)) {
        try {
			LOG(DEBUG) << str() << " readable";
            _recv_from();
        } catch (BadRedisMessage& e) {
            LOG(ERROR) << "Receive bad message from server " << str()
                       << " because: " << e.what()
                       << " dump buffer (before close): "
                       << _buffer.to_string();
            return close_conn();
        }
    }
    _push_to_buffer_set();
    if (poll::event_is_write(events)) {
		LOG(DEBUG) << str() << " writable";
        _output_buffer_set.writev(fd);//返回boolean
    }
    if (_output_buffer_set.empty()) {
		LOG(TRACE) << "_output_buffer_set empty ,set_conn_poll_ro for " << str();
        _proxy->set_conn_poll_ro(this);
    } else {
		LOG(TRACE) << "_output_buffer_set not-empty ,set_conn_poll_rw for " << str();
        _proxy->set_conn_poll_rw(this);
    }
}

void Server::_push_to_buffer_set()
{
    auto now = Clock::now();
    for (util::sref<DataCommand> c: _commands) {
        _sent_commands.push_back(c);
        _output_buffer_set.append(c->buffer);
		LOG(TRACE) << str() << ": push_back _sent_commands/_output_buffer_set.clear _commands";
        c->sent_time = now;
    }
    _commands.clear();
}

void Server::_recv_from()
{
    int n = _buffer.read(fd);
    if (n == 0) {
        throw ConnectionHungUp();
    }
    //LOG(DEBUG) << "Read " << str() << " buffer size " << _buffer.size();
	LOG(TRACE) << "buffer content:" << _buffer.to_string();
    auto responses(split_server_response(_buffer));//返回的responses是vector<responses>

    if (responses.size() > _sent_commands.size()) {
        LOG(ERROR) << "Error on split, expected size: " << _sent_commands.size()
                   << " actual: " << responses.size() << " dump buffer:";
        for (util::sptr<Response> const& rsp: responses) {
            LOG(ERROR) << rsp->get_buffer().to_string();
        }
        LOG(ERROR) << "Rest buffer: " << _buffer.to_string();
        return close_conn();
    }
    LOG(DEBUG) << "responses size=" << responses.size() << ",_sent_commands size=" << _sent_commands.size();
    auto cmd_it = _sent_commands.begin();
    auto now = Clock::now();
    for (util::sptr<Response>& rsp: responses) {
        util::sref<DataCommand> c = *cmd_it++;
        if (c.not_nul()) {
            rsp->rsp_to(c, util::mkref(*this->_proxy));
            c->resp_time = now;
        }
    }
    _sent_commands.erase(_sent_commands.begin(), cmd_it);
    //LOG(TRACE) << "_sent_commands size=" << _sent_commands.size();
}

void Server::push_client_command(util::sref<DataCommand> cmd)
{
	LOG(TRACE) << str() << " push back _commands";
    _commands.push_back(cmd);
    cmd->group->client->add_peer(this);
}

void Server::pop_client(Client* cli)
{
    util::erase_if(
        this->_commands,
        [&](util::sref<DataCommand> cmd)
        {
            return cmd->group->client.is(cli);
        });
    for (util::sref<DataCommand>& cmd: this->_sent_commands) {
        if (cmd.not_nul() && cmd->group->client.is(cli)) {
            cmd.reset();
        }
    }
}

std::vector<util::sref<DataCommand>> Server::deliver_commands()
{
    util::erase_if(
        this->_sent_commands,
        [](util::sref<DataCommand> cmd)
        {
            return cmd.nul();
        });
	LOG(TRACE) << "_commands insert";
    _commands.insert(_commands.end(), _sent_commands.begin(),
                     _sent_commands.end());
    return std::move(_commands);
}

static thread_local std::map<util::Address, Server*> servers_map;
static thread_local std::vector<Server*> servers_pool;
int servers_pool_max_size(8);

static void remove_entry(Server* server)
{
	LOG(TRACE) << "servers_map erase " << server->addr.str();
    ::servers_map.erase(server->addr);
    ::servers_pool.push_back(server);
}

void Server::after_events(std::set<Connection*>&)
{
    if (closed()) {
		LOG(DEBUG) << "Server closed, set slot_map_expired true";
		//LOG(DEBUG) << "sleeping for node recover or slave promoted to master ..." << std::endl;
		//sleep(10);//可等到slave升为master,可能不高效
        _proxy->update_slot_map();
    }
}

std::string Server::str() const
{
    return fmt::format("Server({}@{})", fd, addr.str());
}

//要考虑断了后：
//立刻重启？
//过一段时间重启？
//一直不重启？
void Server::close_conn()
{
    if (!closed()) {
        LOG(INFO) << "Close " << str();
        close();
        _buffer.clear();
        _output_buffer_set.clear();

        for (util::sref<DataCommand> c: _commands) {
            _proxy->retry_move_ask_command_later(c);
        }
        _commands.clear();

        for (util::sref<DataCommand> c: _sent_commands) {
            if (c.nul()) {
				LOG(TRACE) << "";
                continue;
            }
            _proxy->retry_move_ask_command_later(c);
        }
        _sent_commands.clear();

        for (ProxyConnection* conn: attached_long_connections) {
			LOG(TRACE) << "";
            _proxy->inactivate_long_conn(conn);
        }
        attached_long_connections.clear();

        ::remove_entry(this);
    }
}

std::map<util::Address, Server*>::iterator Server::addr_begin()
{
    return ::servers_map.begin();
}

std::map<util::Address, Server*>::iterator Server::addr_end()
{
    return ::servers_map.end();
}

static std::function<void(int, std::vector<util::sref<DataCommand>>&)> on_server_connected(
    [](int, std::vector<util::sref<DataCommand>>&) {});

void Server::_reconnect(util::Address const& addr, Proxy* p)
{
    fd = fctl::new_stream_socket();
    _proxy = p;
    this->addr = addr;

    fctl::set_nonblocking(fd);
    fctl::connect_fd(addr.host, addr.port, fd);
    LOG(INFO) << "Open " << str();
    p->poll_add_rw(this);
    ::on_server_connected(fd, _sent_commands);//还没连接成功就立即调用on_server_connected ？
}

Server* Server::_alloc_server(util::Address const& addr, Proxy* p)
{
    if (servers_pool.empty()) {
		LOG(TRACE) << "servers_pool empty,preallocate 8";
        for (int i = 0; i < 8; ++i) {
            servers_pool.push_back(new Server);
        }
    }
	LOG(TRACE) << "servers_pool size " << servers_pool.size();
    Server* s = servers_pool.back();
    try {
        s->_reconnect(addr, p);
    } catch (IOErrorBase& e) {
        LOG(ERROR) << "Fail to open server " << s->str() << " because " << e.what();
        s->close_conn();//在这里，如果此节点宕机，连接不上，close_conn会进行remove_entry, 向pool中push_back，但是之前也没有弹出啊
    }
    servers_pool.pop_back();
    return s;
}

Server* Server::get_server(util::Address addr, Proxy* p)
{
    auto i = servers_map.find(addr);
    if (i == servers_map.end() || i->second->closed()) {
        Server* s = Server::_alloc_server(addr, p);
        servers_map.insert(std::make_pair(std::move(addr), s));
        return s;
    }else {
		LOG(TRACE) << "server " << addr.str() << " already connected";
	}
    return i->second;
}

static std::string const READONLY_CMD("READONLY\r\n");

void Server::send_readonly_for_each_conn()
{
    ::on_server_connected =
        [](int fd, std::vector<util::sref<DataCommand>>& cmds)
        {
            flush_string(fd, READONLY_CMD);
            cmds.push_back(util::sref<DataCommand>(nullptr));
			LOG(DEBUG) << "flush READONLY_CMD to " << fd;
        };
}
